package com.apps.sdses.flink141.demo;

import com.alibaba.fastjson2.JSON;
import com.apps.sdses.flink141.entity.VehicleDetection;
import com.apps.sdses.flink141.source.VehicleDetectionSource;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author jiwei
 * @description
 * @date 2023/5/31 10:02
 */
public class VehicleDetectionInfoGenerator {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
//        env.setParallelism(5);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        DataStreamSource<VehicleDetection> source = env.addSource(new VehicleDetectionSource());

//        Schema schema = Schema.newBuilder()
//                .column("id", DataTypes.BIGINT())
//                .column("vehicleId", )
//                .build();

//        TableResult execute = tableEnv.fromDataStream(source).execute();


//        tableEnv.executeSql("CREATE TABLE sink_table WITH ('connector' = 'print')\n" +
//                "LIKE source_table (EXCLUDING ALL)");



                source.map(JSON::toJSONString)
                .print();



//        DataStreamSource<Vehicle> source = env.addSource(new VehicleSource());
//        source.map(v -> {
//            return "('" + v.getVehicleId() + "'," + " '" + v.getVehicleType() +"'," +"'" + v.getVehicleBrand() + "','" + v.getLicensePlateNumber() + "')";
//        }).print();

//        source.addSink()


        env.execute();
    }
}
